package com.ds.lens.flink.monitor.monitorjob.function.detector;

import com.ds.lens.data.common.constant.AggregatorEnum;
import com.ds.lens.data.common.constant.CompareOperatorEnum;
import com.ds.lens.data.common.constant.MonitorStateEnum;
import com.ds.lens.data.common.document.DataPoint;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import com.ds.lens.flink.monitor.common.BuilderUtil;
import com.ds.lens.flink.monitor.common.bo.Metric;
import lombok.Builder;
import org.apache.flink.api.java.tuple.Tuple2;
import org.springframework.util.CollectionUtils;

import javax.persistence.Transient;
import java.util.HashSet;
import java.util.Set;


public class ThresholdDetector implements MonitorEventDetector {
    private MonitorStateEnum state;
    private Integer atLeastTimes;
    @Transient
    private MonitorDefinition definition;
    private String metricNameWithAggr;
    private String formula;
    private Double threshold;
    private CompareOperatorEnum operator;

    @Builder
    private ThresholdDetector(CompareOperatorEnum operator, MonitorStateEnum state, Double threshold, Integer atLeastTimes, String metricName, AggregatorEnum aggrEnum) {
        this.state = state;
        this.atLeastTimes = atLeastTimes;
        this.threshold = threshold;
        this.operator = operator;
        this.metricNameWithAggr = RuleExpressionUtil.composeMetricName(metricName, aggrEnum);
        this.formula = RuleExpressionUtil.buildCompareExpression(metricNameWithAggr, operator, threshold);
    }

    @Override
    public MonitorStateEnum getMonitorState() {
        return state;
    }

    @Override
    public MonitorEvent detect(Metric metric) {
        if (CollectionUtils.isEmpty(metric.getValues())) {
            return null;
        }
        int detectCount = 0;
        for (DataPoint point : metric.getValues()) {
            if (point.getState() != null && point.getState().getLevel() > state.getLevel()) {
                detectCount++;
                continue;
            }
            Set<Tuple2<String, Double>> params = new HashSet<>();
            params.add(new Tuple2<>(metricNameWithAggr, point.getValue()));
            boolean result = RuleExpressionUtil.evaluateExpression(formula, params);
            if (result) {
                point.setState(state);
                detectCount++;
            } else {
                point.setState(MonitorStateEnum.OK);
            }
        }
        if (detectCount >= atLeastTimes) {
            return BuilderUtil.buildMonitorEvent(definition, metric, threshold, operator, state);
        } else {
            return null;
        }
    }

    public void setDefinition(MonitorDefinition definition) {
        this.definition = definition;
    }


}
